package org.niit.java;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Random;
import java.util.concurrent.TimeUnit;

public class Data_production {
    public static void main(String[] args) {
        // 创建Kafka生产者
        Properties props = new Properties();
        props.put("bootstrap.servers", "niit202134070927:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> producer = new KafkaProducer<>(props);

        // 定义数据生成器
        Random random = new Random();
        Map<String, String> items = new HashMap<>();
        items.put("apple", "1111");
        items.put("banana", "1112");
        items.put("orange", "1113");
        items.put("phone", "1114");
        items.put("ipad", "1115");
        items.put("pc", "1116");

        while (true) {
            // 生成随机数据
            String itemName = items.keySet().toArray(new String[0])[random.nextInt(items.size())];
            String itemId = items.get(itemName);
            int stock = random.nextInt(100);
            if(stock == 0 || stock < 0){
                stock = stock + 10;
            }
            int sold = random.nextInt(stock);

            String status = random.nextBoolean() ? "U" : "D";

            // 拼接数据
            String data;
            if ("apple".equals(itemName)||"banana".equals(itemName)||"orange".equals(itemName)) {
                data = itemId + "\t" + itemName + "\t" +  "fruit"  + "\t" + stock + "\t" + sold + "\t" + status;
            } else {
                data = itemId + "\t" + itemName + "\t" +  "electronics" + "\t" + stock + "\t" + sold + "\t" + status;
            }


            // 发送数据到Kafka
            producer.send(new ProducerRecord<String, String>("supermarket", itemId, data));

            // 每隔5秒发送一次数据
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}